package com.fafian.publish;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SynchronousBus<S, T> {
	
	private PublishBus bus;
	private Map<BigInteger, SynchronousQueue<ResponseMessage>> responsesQueues;
	protected final Logger logger = LoggerFactory.getLogger(this.getClass());
	private SynchronousListener syncListener;
	private BigInteger busId;

	
	private class SynchronousListener implements Listener<RequestMessage>{
		
		private class ThreadWorker implements Runnable{

			private SynchronousWorker<S,T> worker;
			private T toProcess;
			
			public ThreadWorker(SynchronousWorker<S,T> worker, T toProcess){
				this.worker = worker;
				this.toProcess = toProcess;
			}
			@Override
			public void run() {
				try {
					outputQueue.get(worker.getTargetName()).put(worker.process(toProcess));
				} catch ( SynchronousWorkerException e) {
					logger.error("Exception ocurred while processing {}", e);
				} catch (InterruptedException e) {
					logger.debug("Finished worker before finishing task {}", e);
				}
			
			}
			
		}
		
		private Map<String, List<SynchronousWorker<S,T>>> workers;
		private Map<String, SynchronousQueue<S>> outputQueue;
		private ExecutorService executor;
		
		public SynchronousListener(){
			this.workers = new HashMap<String, List<SynchronousWorker<S,T>>>();
			this.outputQueue =  new  HashMap<String, SynchronousQueue<S>>();
			bus.registerListener(this, RequestMessage.class);
			executor = Executors.newCachedThreadPool();
		}
		
		public synchronized void addWorker(SynchronousWorker<S,T> worker){
			logger.debug("About to add woker for target: \"{}\"",worker.getTargetName());
			if (!workers.containsKey(worker.getTargetName())){
				 List<SynchronousWorker<S,T>> toAdd = new  ArrayList<SynchronousWorker<S,T>>();
				 toAdd.add(worker);
				 workers.put(worker.getTargetName(), toAdd);
			}
			else {
				workers.get(worker.getTargetName()).add(worker);
			}
		}
		
		public boolean isEmpty(){
			return workers.isEmpty();
		}

		@Override
		public void notify(RequestMessage value) {
			if (value.getBusId()== busId){
				@SuppressWarnings("unchecked")
				T obj = (T) value.getContent();
				S result = null;
				if (workers.containsKey(value.getTarget())){
					if (workers.get(value.getTarget()).size()>1){
						SynchronousQueue<S> responseQueue = new SynchronousQueue<S>(); 
						outputQueue.put(value.getTarget(), responseQueue);
						List<Future<?>> listOfFutures = new ArrayList<Future<?>>();
						
						for (SynchronousWorker<S,T> worker :workers.get(value.getTarget())){
							Future<?> future = executor.submit(new ThreadWorker(worker, obj));
							listOfFutures.add(future);
						}
						
						try {
							result = responseQueue.take();
							for (Future<?> future: listOfFutures){
								future.cancel(true);
							}
						} catch (InterruptedException e) {
							logger.error("Error while obtaining response from queue");
						}
					}
					else{
						try {
							result = workers.get(value.getTarget()).get(0).process(obj);
						} catch (SynchronousWorkerException e) {
							logger.error ("Error while processing message {}", e);
						}
					}
				}
				else{
					logger.debug("No workers for target {}", value.getTarget());
				}
				
				ResponseMessage response = new ResponseMessage(result, value.getUniqueId());
				logger.debug("About to publish response {} for id {}", response, response.getUniqueId());
				bus.publish(response);
			}
		}
		

	}
	
	private class ResponseListener implements Listener<ResponseMessage>{
		
		public ResponseListener(){
			
		}
		@Override
		public void notify(ResponseMessage value) {
			SynchronousQueue<ResponseMessage> queue = responsesQueues.get(value.getUniqueId());
			if (queue!=null){
				try {
					queue.put(value);

				} catch (InterruptedException e) {
					queue.clear();
					logger.error("Interrupted while waiting to get information" );
					logger.debug("Detailed exception raised {}",e);

				}
			}
			
		}
	}
	
	
	public SynchronousBus(PublishBus bus){
		ResponseListener rl = new ResponseListener();

		responsesQueues = new  HashMap<BigInteger, SynchronousQueue<ResponseMessage>>();
		if (bus!=null){

			this.bus = bus;
			
		}
		else{
			this.bus = new PublishBus(false);
		}
		this.bus.registerListener(rl, ResponseMessage.class);
		syncListener = new SynchronousListener();
		int n = 16;
        Random r = new Random();
        byte[] b = new byte[n];
        r.nextBytes(b);
        busId = new BigInteger(b);
	}
	
	public SynchronousBus(){
		this(null);
	}
	
	public void registerWorker(SynchronousWorker<S,T> worker){
		syncListener.addWorker(worker);
	}
	
	
	@SuppressWarnings("unchecked")
	public S publish(String target, T obj) throws SynchronousBusException, NoListenerException {
		
		if (bus!=null){
			// queue creation to get response
			if (syncListener.isEmpty()){
				throw new NoListenerException("No registered listener for "+ obj);
			}
			logger.debug("Creating response queue for {}", obj);
			SynchronousQueue<ResponseMessage> responseQueue = new SynchronousQueue<ResponseMessage>();
			logger.debug("Target is: {}", target);
			RequestMessage message = new RequestMessage(target, obj, busId);
			responsesQueues.put(message.getUniqueId(), responseQueue);
			bus.publish(message);
			ResponseMessage response=null;
			try {
				response = responseQueue.take();
			} catch (InterruptedException e) {
				throw new SynchronousBusException(e);
			}
			responsesQueues.remove(message.getUniqueId());
			S toReturn = (S) response.getContent();
			if (toReturn == null){
				throw new SynchronousBusException("Response is null");
			}
			return toReturn;
		}
		else{
			throw new NoListenerException("Bus is null");
		}
	}

}
